package com.renrendoc.network.flow.source;

import com.renrendoc.network.flow.beans.NginxLogEntry;
import com.renrendoc.network.flow.beans.NginxLogEvent;
import com.renrendoc.network.flow.util.PropertiesUtil;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;

import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static com.renrendoc.network.flow.util.Common.convertToDate;
import static com.renrendoc.network.flow.util.Common.parseQueryString;

public class KafkaNginxEntrySource {
    public static void main(String[] args) {

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers(PropertiesUtil.getProperty("kafka.bootstrap_servers"))
                .setTopics("input-topic")
                .setGroupId("my-group")
                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        String log = "[26/Apr/2024:12:12:11 +0800] - 192.168.0.43 \"u=https%3A%2F%2Fdev.renrendoc.com%2Fpaper%2F225639412.html&ru=https%3A%2F%2Fdev.renrendoc.com%2F&tt=%E8%B4%A6%E5%8F%B73619%E7%9A%84%E9%9D%9E%E6%B3%95%E6%B5%8B%E8%AF%95%E6%96%87%E4%BB%B6132.doc&wh=1728x1117&ci=IjxSkCjBFLuQzOzifhtpYZJ9mktsTWY5&si=5rcSwlL1qHw5Y9MW0wNbpCLNnim5MWF4&t=1714100832&rnd=1714100832&et=0&ep=3&cs=%7B%7D&sk=6610cab3e176257b4a7162bc122f243e&v=1.0.0\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36\" 200";

        NginxLogEntry entry = parseLogEntry(log);
        if (entry != null) {
            Map<String, String> urlMap = parseQueryString(entry.getUrl());
            if (200 == entry.getStatus()) {
                NginxLogEvent ngxEvent = new NginxLogEvent();
                ngxEvent.setSk(urlMap.get("sk"))
                        .setSi(urlMap.get("si"))
                        .setCi(urlMap.get("ci"))
                        .setU(urlMap.get("u"))
                        .setRu(urlMap.get("ru"))
                        .setEt(urlMap.get("et"))
                        .setTt(urlMap.get("tt"))
                        .setUa(entry.getUserAgent())
                        .setCs(urlMap.get("cs"))
                        .setV(urlMap.get("v"))
                        .setEp(urlMap.get("ep"))
                        .setWh(urlMap.get("wh"))
                        .setRnd(urlMap.get("aa"))
                        .setIp(entry.getIp())
                        .setTs(convertToDate(entry.getTimestamp()).getTime());

                System.out.println(ngxEvent);
            }
        }


    }

    public static NginxLogEntry parseLogEntry(String log) {
        String regex = "\\[(.*?)\\] - (.*?) \"(.*?)\" \"(.*?)\" (\\d{3})";
        Pattern pattern = Pattern.compile(regex);
        Matcher matcher = pattern.matcher(log);

        if (matcher.find()) {
            String timestamp = matcher.group(1);
            String ip = matcher.group(2);
            String url = matcher.group(3);
            String userAgent = matcher.group(4);
            int status = Integer.parseInt(matcher.group(5));

            return new NginxLogEntry(timestamp, ip, url, userAgent, status);
        }
        return null;
    }
}
